package com.hhf.rrd.product;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.types.Row;

/**
 * 商品优惠券示例
 *           一种补贴发券的场景，希望可以在发券的过程中，对用户领券的上限做控制，
 *      当补贴超过一万元，采用报警，或者别的措施。
 *
 * @author huanghaifeng15
 * @date 2022/2/15 14:59
 **/
@Slf4j
public class ProductCouponJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 1. pdd 发补贴券流水数据
        String createTableSql = "CREATE TABLE source_table (\n"
                + "    id BIGINT,\n" // 补贴券的流水 id
                + "    money BIGINT,\n" // 补贴券的金额
                + "    user_action_time TIMESTAMP(3),\n"
                + "    WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND\n"
                + ") WITH (\n"
                + "  'connector' = 'datagen',\n"
                + "  'rows-per-second' = '1',\n"
                + "  'fields.id.min' = '1',\n"
                + "  'fields.id.max' = '100000',\n"
                + "  'fields.money.min' = '1',\n"
                + "  'fields.money.max' = '100000'\n"
                + ")\n";
        TableResult tableResult = tEnv.executeSql(createTableSql);
        tableResult.print();

        // 2. 计算总计发放补贴券的金额
        String querySql = "SELECT sum(money) as sum_money, -- 补贴券的发放总金额\n"
                + "      count(distinct id) as count_distinct_id\n"
                + "FROM source_table\n"
                + "GROUP BY TUMBLE(user_action_time, INTERVAL '5' SECOND)";
        Table resultTable = tEnv.sqlQuery(querySql);

        // 3. 将金额结果转为 DataStream，然后自定义超过 1w 的报警逻辑
        tEnv.toRetractStream(resultTable, Row.class)
                .flatMap(new RichFlatMapFunction<Tuple2<Boolean, Row>, Row>() {
                    @Override
                    public void flatMap(Tuple2<Boolean, Row> value, Collector<Row> out) throws Exception {
                        Row row = value.f1;
                        assert row != null;
                        assert row.getField(0) != null;
                        long totalMoney = Long.parseLong(String.valueOf(row.getField(0)));
                        if (totalMoney > 10000L) {
                            log.error(value.f1 + " -----> 报警，超过 1w");
                        }
                    }
                });

        env.execute();
    }
}
